package com.king.first.app.order;

import com.king.first.bean.OrderEvent;
import com.king.first.bean.TxEvent;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

public class Project_Order {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> txEventStream  = env.readTextFile("input/ReceiptLog.csv");
        DataStreamSource<String> orderEventStream  = env.readTextFile("input/OrderLog.csv");
        //orderEvent:34729,pay,sd76f87d6,1558430844
        //txEvent:8fdsfae83,alipay,1558430850
        SingleOutputStreamOperator<TxEvent> txEventFlatMapStream = txEventStream.flatMap(new FlatMapFunction<String, TxEvent>() {
            @Override
            public void flatMap(String s, Collector<TxEvent> collector) throws Exception {
                String[] v = s.split(",");
                collector.collect(new TxEvent(v[0], v[1], Long.parseLong(v[2])));
            }
        });
        SingleOutputStreamOperator<OrderEvent> orderEventFLatMapStream = orderEventStream.flatMap(new FlatMapFunction<String, OrderEvent>() {
            @Override
            public void flatMap(String s, Collector<OrderEvent> collector) throws Exception {
                String[] v = s.split(",");
                if ("pay".equals(v[1])){
                    collector.collect(new OrderEvent(Long.parseLong(v[0]), v[1], v[2], Long.parseLong(v[3])));
                }
            }
        });
        ConnectedStreams<TxEvent, OrderEvent> connect = txEventFlatMapStream.connect(orderEventFLatMapStream);

        SingleOutputStreamOperator<Tuple2<OrderEvent, TxEvent>> outputStreamOperator = connect.keyBy(v1 -> v1.getTxId(), v2 -> v2.getTxId()).process(new KeyedCoProcessFunction<String, TxEvent, OrderEvent, Tuple2<OrderEvent, TxEvent>>() {

            private Map<String, TxEvent> txMap = new HashMap();
            private Map<String, OrderEvent> orderMap = new HashMap();

            @Override
            public void processElement1(TxEvent txEvent, KeyedCoProcessFunction<String, TxEvent, OrderEvent, Tuple2<OrderEvent, TxEvent>>.Context context, Collector<Tuple2<OrderEvent, TxEvent>> collector) throws Exception {
                String txId = txEvent.getTxId();
                OrderEvent orderEvent = orderMap.get(txEvent.getTxId());
                if (orderEvent != null) {
                    collector.collect(Tuple2.of(orderEvent, txEvent));
                    orderMap.remove(txId);
                } else {
                    txMap.put(txId, txEvent);
                }
            }

            @Override
            public void processElement2(OrderEvent orderEvent, KeyedCoProcessFunction<String, TxEvent, OrderEvent, Tuple2<OrderEvent, TxEvent>>.Context context, Collector<Tuple2<OrderEvent, TxEvent>> collector) throws Exception {
                String txId = orderEvent.getTxId();
                TxEvent txEvent = txMap.get(txId);
                if (txEvent != null) {
                    collector.collect(Tuple2.of(orderEvent, txEvent));
                    txMap.remove(txId);
                } else {
                    orderMap.put(txId, orderEvent);
                }
            }
        });

        outputStreamOperator.map(new MapFunction<Tuple2<OrderEvent, TxEvent>, String>() {
            @Override
            public String map(Tuple2<OrderEvent, TxEvent> tuple2) throws Exception {
                return tuple2.f0.toString() + "--对账--" + tuple2.f1.toString();
            }
        }).print();

        env.execute();
    }
}
